package Demo1;
import java.util.LinkedList;
import java.util.Map;
import java.util.Queue;
import java.util.Random;
import java.io.*;
import java.net.*;

import backtype.storm.spout.SpoutOutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.IRichSpout;
import backtype.storm.topology.IRichStateSpout;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichSpout;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Values;
import backtype.storm.utils.Utils;


public class DemoSpout implements IRichSpout{
	class RecvCommandThread implements Runnable{
		InputStream _in=null;
		Queue<String> _queue;
		private byte[] _buffer;
		
		public RecvCommandThread(InputStream in,Queue<String> queue) {
			super();
			this._in = in;
			this._queue = queue;
		}

		@Override
		public void run() {
			int n = 0;
			while(true){
				_buffer = new byte[2048];
				try {
					n = _in.read(_buffer);
					String msg = new String(_buffer, 0, n);  
					_queue.add(msg);
				} catch (IOException e) {
					break;
				} 
			}
		}
	}
	
	SpoutOutputCollector _collector;
	static Socket _socket=null;
	InputStream _in=null;
	Queue<String> _queues=new LinkedList<String>();;
	
	Random _rand;
	
	@Override
	public void ack(Object arg0) {}

	@Override
	public void activate() {}

	@Override
	public void deactivate() {}

	@Override
	public void fail(Object arg0) {}

	@Override
	public void close() {
		// TODO Auto-generated method stub
	
		try {
			_socket.close();
			_in.close();
		} catch (UnknownHostException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} 
	}

	@Override
	public void nextTuple() {
		//Utils.sleep(500);
		//String[] sentences = new String[]{"a", "s"};
		//String sentence = sentences[_rand.nextInt(sentences.length)];
		//_collector.emit(new Values(sentence), sentence);
		
		do{
			String sCommand = _queues.poll();
			if (sCommand == null) break;
			_collector.emit(new Values(sCommand), sCommand);
		}while (true);
		//else
		//	Utils.sleep(500);
	}

	@Override
	public void open(Map arg0, TopologyContext arg1, SpoutOutputCollector arg2) {
		// TODO Auto-generated method stub
		_collector = arg2;
		_rand = new Random();
		try{
			_socket = new Socket("192.168.181.124", 8001);
			_in = _socket.getInputStream();  	
			RecvCommandThread ot = new RecvCommandThread(_in, _queues);
			new Thread(ot).start();
		}catch (IOException e){}
	}

	@Override
	public void declareOutputFields(OutputFieldsDeclarer arg0) {
		// TODO Auto-generated method stub
		arg0.declare(new Fields("word"));
	}

	@Override
	public Map<String, Object> getComponentConfiguration() {
		// TODO Auto-generated method stub
		return null;
	}
	
}